package com.jcroke.PC32.utils;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.util.ArrayList;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.CharEncoding;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.jcroke.DACGraph.Pnode;
import com.jcroke.DACGraph.RDFData;
import com.jcroke.DACGraph.DAC.DACClient;
import com.jcroke.PC32.basetransform.Functional;
import com.jcroke.PC32.basetransform.FunctionalContainer;
import com.jcroke.PC32.basetransform.TransformSeqFactory;
import com.jcroke.PC32.datum.dString;
import com.jcroke.PC32.memory.Umem;

/**
 * 
 * @author dmcmillan, jcroke
 *
 */
public class Driver {

	public Driver(File inputDir, File outputDir, String dacHost,
			String dacPort, String dataset) {

		if (!inputDir.isDirectory())
			errorExit(String.format(
					"inputDir directory: %s, is not a directory", inputDir));
		if (!inputDir.canRead())
			errorExit(String.format(
					"inputDir directory: %s, cannot be opened for read",
					inputDir));

		if (!outputDir.exists())
			outputDir.mkdirs();

		if (!outputDir.canWrite())
			errorExit(String.format(
					"output directory: %s, cannot be opened for write",
					outputDir));

		if (!dacPort.matches("^\\d+$")) {
			errorExit(String.format("DAC host port: %s is not a valid port",
					dacPort));
		}

		this._inputFiles = FileUtil.getFilesInDirectory(inputDir, true);
		this._outputDir = outputDir;
		this._dacHost = dacHost;
		this._dacPortAsInt = Integer.parseInt(dacPort);
		this._dataSet = dataset;

		// Create DAC client
		_dac = new DACClient(_dacHost, _dacPortAsInt);

		// Create RDF graphing object
		_rdf = new RDFData(_dac, _dataSet);
	}

	public void run() throws IOException {

		// Build the topological graph.
		_rdf.buildGraph();

		if (_rdf.cyclesDetected()) {
			System.out.println("cycles detected");
			ArrayList<Pnode> cycles = _rdf.getCycles();
			for (Pnode cycle : cycles) {
				System.out.println(cycle.toString());

			}
			errorExit(String.format("Cannot continue."));
		}

		TransformSeqFactory factory = new TransformSeqFactory(_rdf);
		FunctionalContainer tList = null;
		Pnode spout = _rdf.getSpout();
		Pnode outputSerializer = _rdf.getOutputSerializer();
		
		try {
			tList = factory.createTFSList();
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		}

		for (File f : _inputFiles) {
			BufferedReader in = new BufferedReader(new InputStreamReader(
					new FileInputStream(f), CharEncoding.UTF_8));

			File sink = new File(_outputDir, f.getName() + ".out");
			BufferedWriter out = new BufferedWriter(new OutputStreamWriter(
					new FileOutputStream(sink), CharEncoding.UTF_8));
				outputSerializer.setBoWriter(out);
				
			try {
				for (String line = in.readLine(); line != null; line = in
						.readLine()) {

					// Populate spout memory with some input.
					String seedAddr = spout.getOutputSignature().get(0).getSourcePin();
					Umem u = tList.getMemoryObject();
					u.write(spout.getNodeName(), seedAddr, new dString(line));
					
					while (tList.hasFunctional()) {
						Functional transformer = tList.getFunctional();
						transformer.execute();
					}
					
					// Reset the iterator.
					tList.reset();
				}
			} finally {
				System.out.println("[[CLOSE WRITER]]");
				IOUtils.closeQuietly(out);
				IOUtils.closeQuietly(in);
			}
		}
	}

	private static void errorExit(String msg) {
		_logger.info(msg);
		System.exit(-1);
	}

	public static void main(String args[]) throws Exception {
		String input = null;
		String output = null;
		String dataset = null;
		String dacHost = null;
		String dacPort = null;

		for (int i = 0; i < args.length; i++) {
			if (ARG_INPUT.equals(args[i]))
				if (++i < args.length)
					input = args[i];
				else
					errorExit(String.format("missing value for arg: %s",
							ARG_INPUT));

			else if (ARG_OUTPUT.equals(args[i]))
				if (++i < args.length)
					output = args[i];
				else
					errorExit(String.format("missing value for arg: %s",
							ARG_OUTPUT));

			else if (ARG_DAC_PORT.equals(args[i]))
				if (++i < args.length)
					dacPort = args[i];
				else
					errorExit(String.format("missing value for arg: %s",
							ARG_DAC_PORT));
			else if (ARG_DAC_HOST.equals(args[i]))
				if (++i < args.length)
					dacHost = args[i];
				else
					errorExit(String.format("missing value for arg: %s",
							ARG_DAC_HOST));
			else if (ARG_DATASET.equals(args[i]))
				if (++i < args.length)
					dataset = args[i];
				else
					errorExit(String.format("missing value for arg: %s",
							ARG_DATASET));
			else
				errorExit(String.format("unrecognised argument: %s", args[i]));
		}

		if (null == input)
			errorExit(String
					.format("missing mandatory argument: %s", ARG_INPUT));
		if (null == output)
			errorExit(String.format("missing mandatory argument: %s",
					ARG_OUTPUT));
		if (null == dacPort)
			errorExit(String.format("missing mandatory argument: %s",
					ARG_DAC_PORT));
		if (null == dacHost)
			errorExit(String.format("missing mandatory argument: %s",
					ARG_DAC_HOST));
		if (null == dataset)
			errorExit(String.format("missing mandatory argument: %s",
					ARG_DATASET));

		Driver driver = new Driver(new File(input), new File(output), dacHost,
				dacPort, dataset);
		driver.run();
	}

	private static final Logger _logger = LoggerFactory.getLogger(Driver.class);
	private static final String ARG_INPUT = "--inputdir";
	private static final String ARG_OUTPUT = "--outputdir";
	private static final String ARG_DATASET = "--dataset";
	private static final String ARG_DAC_HOST = "--dachost";
	private static final String ARG_DAC_PORT = "--dacport";

	private ArrayList<File> _inputFiles;
	private File _outputDir = null;
	private String _dataSet = null;
	private String _dacHost = null;
	private Integer _dacPortAsInt;
	private RDFData _rdf;
	private DACClient _dac;
}